/*
 * Copyright 2010 Joshua Smith and Nelson Llewellyn
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 *  Unless required by applicable law or agreed to in writing, software
 *  distributed under the License is distributed on an "AS IS" BASIS,
 *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *  See the License for the specific language governing permissions and
 *  limitations under the License.
 */
package norse.server;

import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.CopyOnWriteArraySet;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import norse.msg.ConfigChangeMsgIF;
import norse.msg.MessageIF;
import norse.msg.ScanRequestIF;

/**
 * Singleton that handles distribution of messages throughout the Server.
 */
//TODO: JavaDocs
public final class Postmaster
{
    /**
     * Register a {@code MessageReceiver} for one or more messages.
     *
     * @param _receiver the receiver to register.
     * @param _messages the classes of the messages to register for. The
     * messages must implement either {@code ScanRequestIF} or
     * {@code ConfigChangeMsgIF}.
     *
     * @throws IllegalArgumentException if either parameter is {@code null}. Or
     * invalid.
     */
    public synchronized void registerForMessages(MessageReceiverIF _receiver,
            Class ... _messages)
    {
        if (_receiver == null || _messages == null)
        {
            throw new IllegalArgumentException("Null parameters are invalid");
        }
        else if (_messages.length == 0)
        {
            return;
        }

        //Validate the messages
        for (Class currMsg : _messages)
        {
            Class [] interfaces = currMsg.getInterfaces();

            boolean valid = false;

            for (Class currInterface : interfaces)
            {
                if (currInterface.equals(ScanRequestIF.class)
                        || currInterface.equals(ConfigChangeMsgIF.class))
                {
                    valid = true;
                }
            }

            if (valid == false)
            {
                throw new IllegalArgumentException("All messages must implement"
                        + "ScanRequestIF or ConfigChangeMsgIF. Class \""
                        + currMsg.getName() + "\" does not.");
            }
        }

        //Register for the messages
        for (Class currMsg : _messages)
        {
            Set<MessageReceiverIF> receiverList = messagesMap.get(currMsg);

            //If this message class hasn't been registered yet
            if (receiverList == null)
            {
                /*
                 * Use CopyOnWriteArraySet because modifications to the list
                 * will be rare (if ever) once the program is finished with
                 * initialization
                 */
                receiverList = new CopyOnWriteArraySet<MessageReceiverIF>();

                messagesMap.put(currMsg, receiverList);
            }

            receiverList.add(_receiver);
        }
    }

    /**
     * Deliver a message to the appropriate receiver(s).
     *
     * @param _message the message to deliver.
     *
     * @throws IllegalArgumentException if the parameter is null.
     */
    public void deliverMessage(MessageIF _message)
    {
        if (_message == null)
        {
            throw new IllegalArgumentException("Null parameter is invalid");
        }

        Set<MessageReceiverIF> receivers = messagesMap.get(_message.getClass());
        
        if (receivers == null)
        {
            //TODO: Log as info if we implement server logging.
            return;
        }

        MessageDeliveryTask task = new MessageDeliveryTask(_message, receivers);

        postmasterThreadPool.execute(task);
    }

    /**
     * Get an instance of the postmaster.
     *
     * @return the postmaster
     */
    public static Postmaster getInstance()
    {
        if (thePostmaster == null)
        {
            thePostmaster = new Postmaster();
        }

        return thePostmaster;
    }

    /** The one and only postmaster */
    private static Postmaster thePostmaster = null;

    /**
     * Private constructor to prevent external instantiation.
     */
    private Postmaster()
    {
        messagesMap  = new ConcurrentHashMap<Class, Set<MessageReceiverIF>>();

        postmasterThreadPool = Executors.newFixedThreadPool(THREAD_POOL_SIZE);
    }

    /** A map that links message classes to lists of receivers */
    private ConcurrentMap<Class, Set<MessageReceiverIF>> messagesMap;

    /** The number of threads to allocate to reading and delivering messages */
    private static final int THREAD_POOL_SIZE = 2;

    private ExecutorService postmasterThreadPool;

    private static class MessageDeliveryTask implements Runnable
    {
        public MessageDeliveryTask(MessageIF _message,
                Set<MessageReceiverIF> _receivers)
        {
            message = _message;
            receivers = _receivers;
        }

        private MessageIF message;

        private Set<MessageReceiverIF> receivers;

        @Override
        public void run()
        {

            for (MessageReceiverIF currReceiver : receivers)
            {
                /*
                 * Note: Messages should always be immutable, so no copying
                 * necessary
                 */
                currReceiver.deliverMessage(message);
            }
        }
    }
}
